package com.example.springkafka.prodicer;

import com.alibaba.fastjson.JSON;
import com.example.springkafka.SpringKafkaApplication;
import com.example.springkafka.domain.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

/**
 * @author szh
 */
@RestController
public class HelloProducer {
    public static final String TOPIC_NAME = "topic-test";

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    @PostMapping("/send/{msg}")
    public String sendMsg(@PathVariable("msg") String msg) throws ExecutionException, InterruptedException {
        ListenableFuture<SendResult<Object, Object>> future = kafkaTemplate.send(TOPIC_NAME, msg);

//        User user = new User(null, 12, "张三");
//        String jsonUser = JSON.toJSONString(user);
//        ListenableFuture<SendResult<Object, Object>> send = kafkaTemplate.send(TOPIC_NAME, jsonUser);
        System.out.println("发送结果：" + future.get().toString());
        System.out.println("发送结果：" + future.get().toString());
        return "发送成功";
    }

    @PostMapping("/sendMessage")
    public String sendMsgObj(@RequestBody User user) throws ExecutionException, InterruptedException {
//        ListenableFuture<SendResult<Object, Object>> future = kafkaTemplate.send(SpringKafkaApplication.TOPIC_NAME, msg);
        String jsonUser = JSON.toJSONString(user);
        ListenableFuture<SendResult<Object, Object>> send = kafkaTemplate.send(TOPIC_NAME, jsonUser);
//        System.out.println("发送结果：" + future.get().toString());
        System.out.println("发送结果：" + send.get().toString());


        return "发送成功";
    }

//    @PostMapping("/send/syn/{msg}")
//    public String sendMsgSyn(@PathVariable("msg")String msg) throws ExecutionException, InterruptedException {
//
//        ListenableFuture<SendResult<Object, Object>> future = kafkaTemplate.send(TOPIC_NAME, msg);
//
//        future.addCallback(new ListenableFutureCallback(){
//
//            @Override
//            public void onSuccess(Object result) {
//                System.out.println("发送成功 result = "+result);
//            }
//
//            @Override
//            public void onFailure(Throwable ex) {
//                System.out.println("发送异常");
//                ex.printStackTrace();
//            }
//        });
//
//        return "发送成功";
//    }

}
